package com.yb.kafka.producer.send;

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;

/**
 * 生产者--使用@EnableScheduling注解开启定时任务,kafka建议使用Spring Cloud Stream
 *  ================这里通过类的构造注入对象
 * @author biaoyang
 * @date 2019/9/20 0020
 */
@Slf4j
@Service
@EnableScheduling
@AllArgsConstructor
public class KafkaProducer {
    private final KafkaTemplate kafkaTemplate;
    
    @Scheduled(fixedRate = 10000)
    public void sendMessage(){
        String message = JSONObject.toJSON(new People(LocalDateTime.now().toString(),"jack"+Math.random() , 20, "北京")).toString();
        //推送消息到kafka
        kafkaTemplate.send("myMessage","myKey",message)
                //设置成功或失败的日志信息
                .addCallback(success->log.info("kafka发送信息成功:"+LocalDateTime.now()),fail->log.info("kafka发送信息失败:"+LocalDateTime.now()));
    }

}